package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author:Rzd
 * @Date:2022年10月15日 11:23
 * @Description:
 */
public class DwdTradePayDetailSuc {
    public static void main(String[] args) throws Exception {
        //TODO 1 设置环境
        //1.1设置环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15 * 60 + 5));
        // todo 2 设置检查点
//        // todo 2.1  开启检查点
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        // todo 2.2  设置检查点最小时间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
//        // todo 2.3 job停止后，检查点保存
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        // todo 2.4 设置重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
//        // todo 2.5 检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        // todo 2.6 检查点状态设置
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://mycluster:8020/edu_realtime");
//        // todo 2.7 设置操作hadoop用户
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
        //TODO 3 从kafka读取数据 创建动态表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_pay_suc_group"));
        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                        "data['user_id'] user_id,\n" +
                        "data['order_id'] order_id,\n" +
                        "data['payment_type'] payment_type,\n" +
                        "data['callback_time'] callback_time,\n" +
                        "`proc_time`,\n" +
                        "ts\n" +
                        "from topic_db\n" +
                        "where `table` = 'payment_info'\n"
//                +
//                "and `type` = 'update'\n" +
//                "and data['payment_status']='1602'"
        );
        tableEnv.createTemporaryView("payment_info", paymentInfo);
        //  tableEnv.executeSql("select * from payment_info").print();

        //TODO 5.从下单表中读取下单数据
        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail(\n" +
                " `type` string,\n" +
                " oi_ts string,\n" +
                " `old` map<string,string>,\n" +
                " user_id string,\n" +
                " order_status string,\n" +
                " session_id string,\n" +
                " province_id string,\n" +
                " operate_date string,\n" +
                " operate_time string,\n" +
                " id string,\n" +
                " course_id string,\n" +
                " course_name string,\n" +
                " order_id string,\n" +
                " origin_amount string,\n" +
                " coupon_reduce string,\n" +
                " final_amount string,\n" +
                " create_date string,\n" +
                " create_time string,\n" +
                " od_ts string,\n" +
                " row_op_ts timestamp_ltz(3)\n" +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_pre_process", "dwd_trade_order_per_process"));

        //tableEnv.executeSql("select * from dwd_trade_order_detail").print();
        //TODO 7.将2张表进行连接
        Table resultTable = tableEnv.sqlQuery("" +
                "select\n" +
                "od.id order_detail_id ,\n" +
                "od.order_id ,\n" +
                "od.user_id ,\n" +
                "od.origin_amount ,\n" +
                "od.coupon_reduce ,\n" +
                "od.final_amount ,\n" +
                "od.order_status ,\n" +
                "od.session_id ,\n" +
                "od.province_id ,\n" +
                "od.create_time ,\n" +
                "pi.payment_type ,\n" +
                "pi.callback_time ,\n" +
                "pi.ts \n" +
                "from dwd_trade_order_detail od \n" +
                "join payment_info pi \n" +
                "on od.order_id=pi.order_id\n" +
                "where od.order_status='1002'"
        );
        tableEnv.createTemporaryView("result_table", resultTable);
        // tableEnv.executeSql("select * from result_table").print();


        //TODO 8.创建动态表和要写入的kafka主题进行映射
        tableEnv.executeSql("create table dwd_trade_pay_detail_suc(\n" +
                "order_detail_id string ,\n" +
                "order_id string ,\n" +
                "user_id string ,\n" +
                "origin_amount string ,\n" +
                "coupon_reduce string ,\n" +
                "final_amount string ,\n" +
                "order_status string ,\n" +
                "session_id string ,\n" +
                "province_id string ,\n" +
                "create_time string ,\n" +
                "payment_type string ,\n" +
                "callback_time string ,\n" +
                "ts string ,\n" +
                "primary key(order_detail_id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_detail_suc"));

        //TODO 9.将连接的结果写到kafka主题中
        tableEnv.executeSql("" +
                "insert into dwd_trade_pay_detail_suc select * from result_table");
    }
}
